package com.lib76.socket;

import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.locks.LockSupport;

import org.apache.commons.lang.StringUtils;
import org.jboss.netty.bootstrap.ClientBootstrap;
import org.jboss.netty.bootstrap.ConnectionlessBootstrap;
import org.jboss.netty.channel.Channel;
import org.jboss.netty.channel.ChannelHandlerContext;
import org.jboss.netty.channel.ChannelStateEvent;
import org.jboss.netty.channel.MessageEvent;
import org.jboss.netty.channel.SimpleChannelHandler;

import com.lib76.socket.message.ByteSocketMessage;
import com.lib76.socket.message.JsonSocketMessage;
import com.lib76.socket.message.SpaceSocketMessage;
import com.lib76.socket.model.ReceiveEvent;
import com.lib76.socket.model.SocketChannel;
import com.lib76.socket.model.SocketInfo;
import com.lib76.socket.model.SocketMessage;
import com.lib76.utils.PackageUtils;

/**
 * Socket通讯对象
 * @author 张浩
 * @version 1.0.0
 */
public abstract class SocketObject {
	
	protected SocketInfo info;
	
	protected ConcurrentHashMap<Integer,Thread> waiters;
	
	private Channel channel;
	private ClientBootstrap tcpClient;
	private ConnectionlessBootstrap udpClient;
	
	public SocketObject(SocketInfo info){
		this.info = info;
		this.waiters = new ConcurrentHashMap<Integer,Thread>();
	}
	
	/**
	 * 运行Socket通讯对象
	 * @return	是否运行成功
	 */
	public boolean run(){
		try{
			if(info.getStatusHandler() == null){
				info.setStatusHandler(newInterceptor(IStatusController.class,info.getStatusControlClass()));
			}
			for(String className : info.getInterceptorClasses()){
				IMessageInterceptor temp = newInterceptor(IMessageInterceptor.class,className);
				if(temp != null){
					info.getInterceptors().add(temp);
				}
			}
			if(StringUtils.isNotBlank(info.getInterceptorPath())){
				List<String> classNames = PackageUtils.getClassName(info.getInterceptorPath());
				if(classNames != null && classNames.size() >0){
					for(String className:classNames){
						IMessageInterceptor temp = newInterceptor(IMessageInterceptor.class,className);
						if(temp != null){
							info.getInterceptors().add(temp);
						}
					}
				}
			}
			//启动服务器
			if(info.getLocalPort() != null){
				channel = runServiceSocket();
			}
			//启动客户端
			if(StringUtils.isNotBlank(info.getRemoteIp()) && info.getRemotePort() != null){
				if(SocketTypeEnum.tcp.equals(info.getType())){
					tcpClient = runTcpClientSocket();
				}else if(SocketTypeEnum.tcp.equals(info.getType())){
					udpClient = runUdpClientSocket();
				}
			}
			return true;
		}catch(Exception e){
			e.printStackTrace();
		}
		return false;
	}
	
	/**
	 * 获取无协议的报文
	 * @return
	 */
	public SocketMessage getNoProtocolMessage(){
		if(StringUtils.isNotBlank(this.info.getMessageProtocolVersion())){
			return SocketMessageProtocolManager.createSocketMessage(this.info.getMessageProtocolVersion());
		}else if(StringUtils.isNotBlank(this.info.getMessageProtocolType())){
			if(Constants.MSG_PROTOCOL_TYPE_BYTE.equals(this.info.getMessageProtocolType())){
				return new ByteSocketMessage();
			}else if(Constants.MSG_PROTOCOL_TYPE_JSON.equals(this.info.getMessageProtocolType())){
				return new JsonSocketMessage();
			}else if(Constants.MSG_PROTOCOL_TYPE_SPACE.equals(this.info.getMessageProtocolType())){
				return new SpaceSocketMessage();
			}
		}
		return null;
	}
	
	/**
	 * 获取请求报文
	 * @param protocolCode
	 * @return
	 */
	public SocketMessage getRequestMessage(String protocolCode){
		return SocketMessageProtocolManager.createSocketMessage(this.info.getMessageProtocolVersion(), protocolCode, Constants.MSG_PROTOCOL_DIRECTION_REQUEST);
	}
	
	/**
	 * 获取响应报文
	 * @param protocolCode
	 * @return
	 */
	public SocketMessage getResponseMessage(String protocolCode){
		return SocketMessageProtocolManager.createSocketMessage(this.info.getMessageProtocolVersion(), protocolCode, Constants.MSG_PROTOCOL_DIRECTION_RESPONSE);
	}
	
	/**
	 * 当Socket对象为服务器时，获取服务器通讯通道
	 * @return
	 */
	public SocketChannel getServerChannel(){
		return new SocketChannel(channel);
	}
	
	/**
	 * 当Socket对象为客户端时，获取当前客户端通讯通道
	 * @return
	 */
	public SocketChannel getClientChannel(){
		if(tcpClient != null){
			Channel channel = tcpClient.connect(this.info.getRemoteSocketAddress()).getChannel();
			waiters.put(channel.getId(), Thread.currentThread());
			LockSupport.park(channel);
			return new SocketChannel(channel);
		}else if(udpClient != null){
			Channel channel = udpClient.connect(this.info.getRemoteSocketAddress()).getChannel();
			return new SocketChannel(channel);
		}
		return null;
	}

	protected abstract Channel runServiceSocket() throws Exception;
	
	protected abstract ClientBootstrap runTcpClientSocket() throws Exception;
	
	protected abstract ConnectionlessBootstrap runUdpClientSocket() throws Exception;
	
	private<T> T newInterceptor(Class<T> interfaceClassName,String interceptorClass){
		try{
			if(StringUtils.isNotBlank(interceptorClass)){
				Class<?> className = Class.forName(interceptorClass);
				for(Class<?> item : className.getInterfaces()){
					if(item.equals(interfaceClassName)){
						return interfaceClassName.cast(className.newInstance());
					}
				}
			}
		}catch(Exception e){
			e.printStackTrace();
		}
		return null;
	}
	
	public class Handler extends SimpleChannelHandler{
		
        public void channelConnected(ChannelHandlerContext ctx, ChannelStateEvent e) throws Exception {
        	SocketChannel channel = new SocketChannel(e.getChannel());
        	SocketChannelManager.addSocketChannel(channel);
        	LockSupport.unpark(waiters.remove(channel.getId()));
        	if(info.getStatusHandler() != null){
        		info.getStatusHandler().channelConnected(channel);
        	}
        }

        public void messageReceived(ChannelHandlerContext ctx, MessageEvent e) throws Exception {
        	SocketObject socketObject = CommunicationServer.getSocketObject(info.getId());
        	//System.out.println(e.getMessage());
        	SocketMessage message = null;
        	if(StringUtils.isNotBlank(info.getMessageProtocolVersion())){
        		message = SocketMessageProtocolManager.createSocketMessage(info.getMessageProtocolVersion(),e.getMessage());
        	}else{
        		message = getNoProtocolMessage();
        		message.readOriginalMessage(e.getMessage());
        	}
        	ReceiveEvent event = new ReceiveEvent(socketObject,new SocketChannel(e.getChannel()),message);
        	for(IMessageInterceptor interceptor : info.getInterceptors()){
        		if(interceptor.intercept(event)){
        			return;
        		}
        	}
        }
        
        public void channelClosed(ChannelHandlerContext ctx, ChannelStateEvent e) throws Exception {
        	SocketChannel channel = new SocketChannel(e.getChannel());
        	if(info.getStatusHandler() != null){
        		info.getStatusHandler().channelClosed(channel);
        	}
        	SocketChannelManager.removeSocketChannel(channel.getId());
        }

        public void channelDisconnected(ChannelHandlerContext ctx, ChannelStateEvent e) throws Exception {
        	if(info.getStatusHandler() != null){
        		info.getStatusHandler().disconnected();
        	}
        }  
   }
}
